package tbdp.tas.hbase.impl;

import java.util.Map.Entry;

import org.apache.commons.collections.MapUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.HTablePool;
import org.apache.hadoop.hbase.util.PoolMap.PoolType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import tbdp.tas.hbase.HBaseConnection;
import tbdp.tas.hbase.HbaseConnector;


/**
 * hbase 链接池实现
 * @author xbkaishui
 * @version $Id: HbaseConnectorImpl.java, v 0.1 2016-5-25 下午9:10:32 xbkaishui Exp $
 */
public class HbaseConnectorImpl implements HbaseConnector {

    private Logger             log                                      = LoggerFactory
                                                                            .getLogger(HbaseConnectorImpl.class);

    /** Name of ZooKeeper quorum configuration parameter. */
    public static final String ZOOKEEPER_QUORUM                         = "hbase.zookeeper.quorum";

    /** Parameter name for the root dir in ZK for this cluster */
    public static final String ZOOKEEPER_ZNODE_PARENT                   = "zookeeper.znode.parent";

    /** hbase htable pool size key name */
    public static final String HTABLE_POOL_SIZE                         = "hbase.htable.poolsize";

    /** timeout for each RPC */
    public static String       HBASE_RPC_TIMEOUT                        = "hbase.rpc.timeout";

    public static String       HBASE_CLIENT_PAUSE                       = "hbase.client.pause";

    public static String       HBASE_CLIENT_RETRIES_NUMBER              = "hbase.client.retries.number";

    public static String       HBASE_IPC_CLIENT_TCPNODELAY              = "hbase.ipc.client.tcpnodelay";

    public static String       HBASE_IPC_PING_INTERVAL                  = "ipc.ping.interval";

    public static String       HBASE_CLIENT_SCANNER_CACHING             = "hbase.client.scanner.caching";

    public static String       HBASE_ZOOKEEPER_SESSION_TIMEOUT          = "zookeeper.session.timeout";

    public static String       HBASE_IPC_CLIENT_CONNECTION_MAXIDLE_TIME = "hbase.ipc.client.connection.maxidletime";

    /** htable连接池 **/
    private HTablePool         tablePool;
    /**hbase配置*/
    private Configuration      cfg;

    /**必须填写的两项属性*/
    private String             hbaseZookeeperQuorum;
    private String             hbaseZookeeperZnode;
    /**选填的属性*/
    private int                hbaseTablePoolSize                       = 50;
    private String             hbaseClientPause                         = "100";
    private String             hbaseClientRetriesNumber                 = "3";
    private String             hbaseIpcClientTcpnodelay                 = "true";
    private String             ipcPingInterval                          = "3000";
    private String             hbaseRpcTimeout                          = "20000";
    private String             hbaseClientScannerCaching                = "20";
    private String             zookeeperSessionTimeout                  = "90000";
    public static String       hbaseIpcClientConnectionMaxidleTime      = "36000000";

    private HBaseConnection    conn;

    /**
     * @param conn
     */
    public HbaseConnectorImpl(HBaseConnection conn) {
        log.info("hbase conn " + conn);
        this.conn = conn;
    }

    /**
     * 初始化配置
     * 
     */
    private void initConfig() {
        this.hbaseZookeeperQuorum = conn.getZk();
        this.hbaseZookeeperZnode = conn.getZnode();

        // 设置hbase配置参数
        cfg.set(ZOOKEEPER_QUORUM, hbaseZookeeperQuorum);
        cfg.set(ZOOKEEPER_ZNODE_PARENT, hbaseZookeeperZnode);

        cfg.set(HBASE_CLIENT_PAUSE, hbaseClientPause);
        cfg.set(HBASE_CLIENT_RETRIES_NUMBER, hbaseClientRetriesNumber);
        cfg.set(HBASE_IPC_CLIENT_TCPNODELAY, hbaseIpcClientTcpnodelay);
        cfg.set(HBASE_IPC_PING_INTERVAL, ipcPingInterval);
        cfg.set(HBASE_RPC_TIMEOUT, hbaseRpcTimeout);
        cfg.set(HBASE_CLIENT_SCANNER_CACHING, hbaseClientScannerCaching);
        cfg.set(HBASE_ZOOKEEPER_SESSION_TIMEOUT, zookeeperSessionTimeout);
        cfg.set(HBASE_IPC_CLIENT_CONNECTION_MAXIDLE_TIME, hbaseIpcClientConnectionMaxidleTime);

        /**初始化额外配置*/
        if (MapUtils.isNotEmpty(conn.getConf())) {
            for (Entry<String, String> entry : conn.getConf().entrySet()) {
                cfg.set(entry.getKey(), entry.getValue());
            }
        }
    }

    /** 
     */
    public void init() {
        long beginTime = System.currentTimeMillis();
        try {
            // 环境 问题
            System.setProperty("javax.xml.parsers.DocumentBuilderFactory",
                "com.sun.org.apache.xerces.internal.jaxp.DocumentBuilderFactoryImpl");
            System.setProperty("javax.xml.parsers.SAXParserFactory",
                "com.sun.org.apache.xerces.internal.jaxp.SAXParserFactoryImpl");

            cfg = new Configuration();
            // 初始化配置
            initConfig();
            //table 设置成 thread local 类型
            tablePool = new HTablePool(cfg,hbaseTablePoolSize,PoolType.ThreadLocal);

            long endTime = System.currentTimeMillis();
            if (log.isWarnEnabled()) {
                log.warn("Hbase连接器初始化成功! hbaseZookeeper:" + hbaseZookeeperQuorum + "Poolsize:"
                         + hbaseTablePoolSize + ",ellapsedTime=" + (endTime - beginTime));
            }
        } catch (Exception e) {
            log.error("Hbase连接器初始化过程异常!", e);
            throw new RuntimeException(e);
        }
    }

    public HTableInterface getTable(String tableName) {
        return tablePool.getTable(tableName);
    }

    public Configuration getConfiguration() {
        return cfg;
    }

}
